package channel

import (
	"micro-rulego/internal/broker"
	"sync"
)

type channelDown struct {
	active bool
	topic  string
	broker broker.Broker
	suber  broker.Subscriber
	chd    chan []byte
	exit   chan struct{}
	m      sync.Mutex
}

func (cd *channelDown) Put(b []byte) error {
	return ErrChannelNotSupport
}

func (cd *channelDown) Get() ([]byte, error) {
	select {
	case b := <-cd.chd:
		return b, nil
	case <-cd.exit:
		return nil, ErrChannelInvalid
	}
}

func (cd *channelDown) Topic() string {
	return cd.topic
}

func (cd *channelDown) Type() bool {
	return true
}

func (cd *channelDown) Start() error {
	cd.m.Lock()
	defer cd.m.Unlock()

	if cd.active {
		return ErrChannelBusy
	}

	if cd.exit == nil {
		cd.exit = make(chan struct{})
	}
	if cd.chd == nil {
		cd.chd = make(chan []byte, 10)
	}

	Subscriber, err := cd.broker.Subscribe(cd.topic, func(e broker.Event) error {
		select {
		case cd.chd <- e.Message():
			return nil
		default:
			return ErrChannelBusy
		}
	})
	if err != nil {
		// 如果订阅失败，关闭已创建的 channels
		close(cd.exit)
		cd.exit = nil
		close(cd.chd)
		cd.chd = nil
		return err
	}

	cd.active = true
	cd.suber = Subscriber
	return nil
}

func (cd *channelDown) Stop() error {
	cd.m.Lock()
	defer cd.m.Unlock()

	if !cd.active {
		return nil
	}

	// 先发送退出信号
	if cd.exit != nil {
		close(cd.exit)
		cd.exit = nil // 清空引用
	}

	// 取消订阅
	if cd.suber != nil {
		if err := cd.suber.Unsubscribe(); err != nil {
			return err
		}
		cd.suber = nil // 清空引用
	}

	// 最后关闭数据channel
	if cd.chd != nil {
		close(cd.chd)
		cd.chd = nil // 清空引用
	}

	cd.active = false
	return nil
}

type channelUp struct {
	topic  string
	broker broker.Broker
}

func (cu *channelUp) Put(b []byte) error {
	return cu.broker.Publish(cu.topic, b)
}

func (cu *channelUp) Get() ([]byte, error) {
	return nil, ErrChannelNotSupport
}

func (cu *channelUp) Topic() string {
	return cu.topic
}

func (cu *channelUp) Type() bool {
	return false
}

func (cu *channelUp) Start() error {
	return nil
}

func (cu *channelUp) Stop() error {
	return nil
}

func NewChannelBroker(topic string, isDown bool, broker broker.Broker) Channel {
	if isDown {
		return &channelDown{
			active: false,
			topic:  topic,
			broker: broker,
		}
	} else {
		return &channelUp{
			topic:  topic,
			broker: broker,
		}
	}
}
